package com.gjy.disruptor.db2;

import com.lmax.disruptor.EventHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-10-24 20:39:58
 */
public class UserAttentionConsumer implements EventHandler<UserAttentionEntity> {
    private static final Logger log = LoggerFactory.getLogger(UserAttentionConsumer.class);

    private static final String SQL = "insert into user_attention_bak (ua_id, user_id, user_attention_id) values (?,?,?)";
    private static final Integer BATCH_SIZE = 5000;
    private static final List<UserAttentionEntity> sqlList = new CopyOnWriteArrayList<>();

    private final Connection connection;

    public UserAttentionConsumer(Connection connection) {
        this.connection = connection;
        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
            connection.setAutoCommit(false);
        } catch (SQLException | ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void onEvent(UserAttentionEntity event, long sequence, boolean endOfBatch) throws Exception {
        sqlList.add(event);
        if (endOfBatch || sqlList.size() >= BATCH_SIZE) {
            processSql();
        }
    }

    private void processSql() {
        try (PreparedStatement statement = connection.prepareStatement(SQL)) {
            for (UserAttentionEntity s : sqlList) {
                statement.setObject(1, s.getUaId());
                statement.setObject(2, s.getUserId());
                statement.setObject(3, s.getUserAttentionId());
                statement.addBatch();
            }
            statement.executeBatch();
            connection.commit();
            statement.clearBatch();
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            log.info("处理了: {}", sqlList.size());
            sqlList.clear();
        }
    }

}
